package com.pw.study.flink.cep;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.util.List;
import java.util.Map;

/**
 * @Author: linux_future
 * @since: 2022/3/20
 **/
public class ExerciseCep extends BaseCEP {
    public static void main(String[] args) {
        new ExerciseCep().init();
    }

    @Override
    protected void handler(StreamExecutionEnvironment env, SingleOutputStreamOperator<JSONObject> stream) {
        //stream.print();
        SingleOutputStreamOperator<JSONObject> data = stream
                .map((MapFunction<JSONObject, JSONObject>) value -> value.getJSONObject("common"));
        //data.print();
        //定义模式
        Pattern<JSONObject, JSONObject> pattern = Pattern.<JSONObject>begin("start")
                .where(new SimpleCondition<JSONObject>() {
                    @Override
                    public boolean filter(JSONObject value) throws Exception {
                        String os = value.getString("os");
                        System.out.println("os: " + os);
                        return "Appstore".equals(value.getString("ch"));
                    }
                });
        pattern.times(2);
        //绑定模式
        PatternStream<JSONObject> ps = CEP.pattern(data, pattern);
        //获取匹配结果
        ps.select(new PatternSelectFunction<JSONObject, String>() {
            @Override
            public String select(Map<String, List<JSONObject>> value) throws Exception {
                System.out.println("====value");
                return value.toString();
            }
        }).print();

    }
}
